package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	//	"bytes"

	"bytes"
	"fmt"
	"math/rand"
	"sort"
	"sync"
	"sync/atomic"
	"time"

	//	"6.824/labgob"
	"6.824/labgob"
	"6.824/labrpc"
)

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

//
// A Go object implementing a single Raft peer.
//
type any = interface{}

const (
	Follower            int           = 1
	Candidate           int           = 2
	Leader              int           = 3
	electionDuationMin  time.Duration = time.Duration(500 * time.Millisecond)
	electionDuationMax  time.Duration = time.Duration(1000 * time.Millisecond)
	heartsbeatsDurtaion time.Duration = time.Duration(50 * time.Millisecond)
	startnewtermDuation time.Duration = time.Duration(1 * time.Second)
)

type Log struct {
	Term int
	Log  interface{}
}
type ST_snapShot struct {
	index    int
	term     int
	snapshot []byte
}
type Raft struct {
	mu            sync.Mutex // Lock to protect shared access to this peer's state
	muApplyChange sync.Mutex
	LastLockInfo  string
	peers         []*labrpc.ClientEnd // RPC end points of all peers
	persister     *Persister          // Object to hold this peer's persisted state
	me            int                 // this peer's index into peers[]
	dead          int32               // set by Kill()

	status      int
	currentTerm int   /*Persistent*/
	votedFor    int   /*Persistent*/
	log         []Log /*Persistent*/
	snapShot    ST_snapShot
	commitIndex int
	lastApplied int
	nextIndex   []int
	matchIndex  []int
	lastOpTime  time.Time
	applyCh     chan ApplyMsg
	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) Lock(debugInfo ...any) {
	for count := 0; ; count++ {
		if count == 100 {
			rf.PrintlnAlreadyLock(direction{-1, -1}, "Dead Lock, Prev Info is ", rf.LastLockInfo, "Now is ", fmt.Sprint(debugInfo...))
		}
		if rf.mu.TryLock() == false {
			time.Sleep(time.Duration(time.Millisecond * 1))
		} else {
			break
		}
	}
	rf.LastLockInfo = fmt.Sprint(debugInfo...)
	// rf.mu.Lock()
}
func (rf *Raft) Unlock(debugInfo ...any) {
	rf.LastLockInfo = ""
	rf.mu.Unlock()
}
func (rf *Raft) GetState() (int, bool) {

	var term int
	var isleader bool
	rf.Lock("GetState")
	defer rf.Unlock()
	term = rf.currentTerm
	isleader = rf.status == Leader
	// Your code here (2A).
	return term, isleader
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	rf.Lock()
	e.Encode(rf.currentTerm)
	e.Encode(rf.votedFor)
	e.Encode(rf.log)
	// e.Encode(rf.snapShot.index)
	// e.Encode(rf.snapShot.term)
	data := w.Bytes()
	rf.persister.SaveStateAndSnapshot(data, rf.snapShot.snapshot)
	// rf.PrintlnAlreadyLock(direction{-1, -1}, "Save data log = ", rf.log, "term = ", rf.currentTerm, "votedfor = ", rf.votedFor)
	rf.Unlock()
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}

	// Your code here (2C).
	// Example:
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var currentTerm int /*Persistent*/
	var votedFor int    /*Persistent*/
	var log []Log       /*Persistent*/
	// var snapshotindex int
	// var snapshotterm int
	if d.Decode(&currentTerm) != nil ||
		d.Decode(&votedFor) != nil ||
		d.Decode(&log) != nil {

	} else {
		rf.Lock()
		rf.currentTerm = currentTerm
		rf.votedFor = votedFor
		rf.log = log
		// rf.snapShot.index = snapshotindex
		// rf.snapShot.term = snapshotterm
		rf.Unlock()
	}
	rf.PrintlnAlreadyLock(direction{-1, -1}, "Read data back, log = ", rf.log, "term = ", rf.currentTerm, "votedfor = ", rf.votedFor)
}

//
// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
//
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).

	return true
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).
	rf.Lock("Snapshot")
	// if rf.status != Leader {
	// 	fmt.Println(rf.me, "Run Snapshot, but not leader")
	// 	rf.Unlock()
	// 	return
	// }
	// if index <= rf.snapShot.index {
	// 	fmt.Println(rf.me, "Run Snapshot, but old command")
	// 	rf.Unlock()
	// 	return
	// }

	LastLog := *rf.GetLog(index)
	rf.log = append([]Log(nil), rf.GetLogRange(index+1, rf.GetLogLen())...)
	rf.snapShot.index = index
	rf.snapShot.term = LastLog.Term

	// rf.snapShot.snapshot = make([]byte, 0, len(snapshot))
	// copy(rf.snapShot.snapshot, snapshot)
	rf.snapShot.snapshot = snapshot
	rf.Unlock()
	rf.persist()
	// fmt.Println(rf.me, "snapshot succeed", rf.snapShot.index, rf.snapShot.term, snapshot, "AfterCopy", rf.snapShot, rf.log)
	// rf.PrintAllLogs()
	// for i, p := range rf.peers {
	// 	if i != rf.me {
	// 		// go rf.LeaderInstallSnapshot(i, p)
	// 	}
	// }
}

type InstallSnapshotArgs struct {
	Term              int
	LeaderID          int
	LastIncludedIndex int
	LastIncludedTerm  int
	/*
		The Requirement is :
		Don't implement Figure 13's offset mechanism for splitting up the snapshot.
		so, disable the offect and done field
	*/
	// offset int
	Data []byte
	// done   bool
}
type InstallSnapshotReply struct {
	Term int
}

func (rf *Raft) InstallSnapshot(args *InstallSnapshotArgs, reply *InstallSnapshotReply) {
	// fmt.Println(rf.me, "Intall snapshot", args.LastIncludedIndex, args.LastIncludedTerm, args.LeaderID, args.Term)
	// fmt.Println(rf.me, "Intall snapshot", *args)
	SkipInstall := rf.snapShot.index == args.LastIncludedIndex
	// fmt.Println(rf.me, "Situation:", rf.snapShot.index, rf.snapShot.term, rf.currentTerm, rf.log)
	rf.Lock("InstallSnapshot")
	if rf.commitIndex < args.LastIncludedIndex {
		rf.commitIndex = args.LastIncludedIndex
		rf.Unlock()
		// rf.ApplyChanges(rf.commitIndex)
		rf.Lock()
	}

	reply.Term = rf.currentTerm
	if args.Term < rf.currentTerm {
		rf.Unlock()
		return
	}
	if rf.snapShot.index > args.LastIncludedIndex {
		rf.Unlock()
		return
	}
	if rf.snapShot.index == args.LastIncludedIndex {
		if rf.snapShot.term == args.LastIncludedTerm {
			rf.Unlock()
			return
		} else {
			// fmt.Println("Unexpected Error! Snapshot conflict")
			rf.Unlock()
			return
		}
	}
	rf.snapShot.snapshot = args.Data
	beforeSnapIndex := rf.snapShot.index
	rf.snapShot.term = args.LastIncludedTerm
	deltaLogNum := args.LastIncludedIndex - beforeSnapIndex
	if args.LastIncludedIndex <= rf.GetLogLen() {
		lastLogTerm := rf.GetLog(args.LastIncludedIndex).Term
		if lastLogTerm == args.LastIncludedTerm {
			rf.log = append([]Log(nil), rf.log[deltaLogNum:]...)
		} else {
			rf.log = make([]Log, 0)
		}
	} else {
		rf.log = make([]Log, 0)
	}
	rf.snapShot.index = args.LastIncludedIndex

	if rf.currentTerm < args.Term {
		rf.SwitchToFollower(-1, args.Term)
	}
	rf.Unlock()

	// Send apply msg
	if !SkipInstall {
		rf.ApplySnapshot(args)
	}
	rf.muApplyChange.Lock()
	rf.lastApplied = max(rf.lastApplied, args.LastIncludedIndex)
	rf.muApplyChange.Unlock()

	//persist
	rf.persist()
	rf.Lock()
	// fmt.Println(rf.me, "After Install,", "Situation:", rf.snapShot.index, rf.snapShot.term, rf.currentTerm, rf.log)
	// rf.PrintAllLogs()
	rf.Unlock()
}
func (rf *Raft) ApplySnapshot(args *InstallSnapshotArgs) {
	msg := ApplyMsg{false, nil, 0, true, args.Data, args.LastIncludedTerm, args.LastIncludedIndex}
	// fmt.Println(rf.me, "Send Snapshot MSG:", msg)
	rf.applyCh <- msg

}
func (rf *Raft) LeaderInstallSnapshot(id int) {
	if id == rf.me {
		return
	}
	peer := rf.peers[id]
	rf.Lock()
	args := InstallSnapshotArgs{rf.currentTerm, rf.me, rf.snapShot.index, rf.snapShot.term, rf.snapShot.snapshot}
	rf.Unlock()
	reply := InstallSnapshotReply{}
	ok := peer.Call("Raft.InstallSnapshot", &args, &reply)
	if ok {
		rf.Lock("leaderSendAppendEntries 2")
		if reply.Term == rf.currentTerm {
			rf.matchIndex[id] = args.LastIncludedIndex
			rf.nextIndex[id] = rf.matchIndex[id] + 1
		} else if reply.Term < rf.currentTerm {
			// fmt.Println("Unexpected Error of calling InstallSnapshot")
		} else {
			rf.SwitchToFollower(-1, reply.Term)
			return
		}
		rf.Unlock()
	} else {
		rf.Println(direction{0, id}, "call Raft.InstallSnapshot failed")
		return
	}
}

type AppendEntriesArgs struct {
	Term         int
	LeaderID     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []Log
	LeaderCommit int
}
type AppendEntriesReply struct {
	Term             int
	Success          bool
	CorrectPrevIndex int
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	reply.CorrectPrevIndex = -1
	reply.Success = true
	rf.Lock("AppendEntries")
	var CheckPrevLog bool = true
	var prevLog Log
	if args.Term < rf.currentTerm {
		/*Not Vaild Term, Skip Directly*/
		reply.Term = rf.currentTerm
		reply.Success = false
		rf.Unlock()
		return
	}
	rf.lastOpTime = time.Now()
	if rf.GetLogLen() < args.PrevLogIndex {
		/*PrevLogIndex out of range Need Decrease PrevLogIndex*/
		CheckPrevLog = false
	} else if args.PrevLogIndex != 0 && args.PrevLogIndex <= rf.GetLogLen() {
		prevLog = *rf.GetLog(args.PrevLogIndex)
		if prevLog.Term != args.PrevLogTerm {
			CheckPrevLog = false
		}
	}
	if CheckPrevLog == false {
		reply.Success = false
		count := 0
		for reply.CorrectPrevIndex = min(rf.GetLogLen(), args.PrevLogIndex); reply.CorrectPrevIndex > 0; reply.CorrectPrevIndex-- {
			if reply.CorrectPrevIndex <= rf.snapShot.index {
				break
			}
			if rf.GetLog(reply.CorrectPrevIndex).Term < min(args.PrevLogTerm, prevLog.Term) {
				break
			} else {
				count++
			}
			if count > 100 {
				break
			}
		}
		reply.CorrectPrevIndex++
		reply.Term = rf.currentTerm
	} else {
		if args.PrevLogIndex+len(args.Entries) < rf.GetLogLen() {
			if len(args.Entries) > 0 && args.Entries[len(args.Entries)-1].Term == rf.GetLog(args.PrevLogIndex+len(args.Entries)).Term {
				/*Log match but have entries, Old Message, Drop*/
				rf.Unlock()
				return
			}
		}
		rf.TruncLog(args.PrevLogIndex)
		rf.log = append(rf.log, args.Entries...)
	}

	status := rf.status
	term := rf.currentTerm

	if term < args.Term {
		rf.SwitchToFollower(-1, args.Term)
	} else if term == args.Term && status != Leader {
		if status != Follower {
			rf.SwitchToFollower(-1, args.Term)
		}
	} else if term == args.Term && status == Leader {
	} else {
		/*term > args.Term*/
		reply.Term = term
		reply.Success = false
	}

	if reply.Success == true && args.LeaderCommit > rf.commitIndex {
		rf.commitIndex = min(args.LeaderCommit, rf.GetLogLen())
		go rf.ApplyChanges(rf.commitIndex)
	} else {
	}
	// rf.PrintlnAlreadyLock(direction{1, args.LeaderID}, "Complete AppendEntries, args = ", *args, "reply = ", *reply, "rf.commitIndex = ", rf.commitIndex, "rf.log = ", rf.log)
	rf.PrintlnAlreadyLock(direction{1, args.LeaderID}, "Complete AppendEntries, args = ", *args, "reply = ", *reply, "rf.commitIndex = ", rf.commitIndex, "rf log len = ", rf.GetLogLen())
	rf.Unlock()
	rf.persist()

	return
}
func min(a int, b int) int {
	if a < b {
		return a
	}
	return b
}
func max(a int, b int) int {
	if a < b {
		return b
	}
	return a
}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateID  int
	LastLogIndex int
	LastLogTerm  int
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.Lock("RequestVote")
	currentTerm := rf.currentTerm
	votedFor := rf.votedFor
	LastLogIndex := rf.GetLogLen()
	var LastLogTerm int = 0
	if LastLogIndex != 0 {
		LastLogTerm = rf.GetLog(LastLogIndex).Term
	}
	// rf.PrintlnAlreadyLock(direction{0, args.CandidateID}, "Receive RequestVote, args = ", *args, "currentTerm = ", currentTerm, "votedFor = ", votedFor, "LastLogIndex = ", LastLogIndex, "LastLogTerm = ", LastLogTerm)
	if args.Term > rf.currentTerm {
		rf.SwitchToFollower(-1, args.Term)
		votedFor = rf.votedFor
	}

	if args.Term < currentTerm {
		reply.VoteGranted = false
		reply.Term = currentTerm
		/*The come in have a newer term ID, change the term ID*/
		rf.SwitchToFollower(-1, args.Term)
	} else if votedFor == -1 || votedFor == args.CandidateID {
		if args.LastLogTerm > LastLogTerm || (args.LastLogIndex >= LastLogIndex && args.LastLogTerm == LastLogTerm) {
			reply.VoteGranted = true
		}
	}
	if reply.VoteGranted {
		rf.SwitchToFollower(args.CandidateID, args.Term)
	} else {
		reply.Term = currentTerm
	}
	// rf.PrintlnAlreadyLock(direction{1, args.CandidateID}, "reply = ", *reply)
	rf.Unlock()
	rf.persist()
}
func (rf *Raft) SwitchToFollower(votedFor int, currentTerm int) {
	// rf.Lock()
	// defer rf.Unlock()
	// rf.PrintlnAlreadyLock(direction{-1, -1}, "SwitchToFollower")
	rf.votedFor = votedFor
	rf.currentTerm = currentTerm
	rf.status = Follower
	if votedFor != -1 {
		rf.lastOpTime = time.Now()
	}
}
func (rf *Raft) SwitchToCandicate() {
	var a []RequestVoteArgs = make([]RequestVoteArgs, len(rf.peers))
	// var r []RequestVoteReply = make([]RequestVoteReply, len(rf.peers))
	rf.Lock("SwitchToCandicate")
	// rf.PrintlnAlreadyLock(direction{-1, -1}, "SwitchToCandicate")
	rf.lastOpTime = time.Now()
	rf.status = Candidate
	rf.currentTerm++
	rf.votedFor = rf.me
	var LastLogIndex, LastLogTerm int
	if rf.GetLogLen() > 0 {
		LastLogIndex = rf.GetLogLen()
		LastLogTerm = rf.GetLog(LastLogIndex).Term
	} else {
		LastLogIndex = 0
		LastLogTerm = 0
	}

	for id, _ := range a {
		a[id] = RequestVoteArgs{rf.currentTerm, rf.me, LastLogIndex, LastLogTerm}
	}
	rf.Unlock()
	rf.persist()
	var NotCompleteCount int = len(rf.peers) - 1
	var AcceptCount int = 1
	var AcceptList []int
	AcceptList = append(AcceptList, rf.me)
	var m sync.Mutex
	for i := 0; i < len(rf.peers); i++ {
		if i != rf.me {
			go func(args *RequestVoteArgs, id int) {
				var reply RequestVoteReply = RequestVoteReply{-1, false}
				ok := rf.peers[id].Call("Raft.RequestVote", args, &reply)
				m.Lock()
				if ok == false {
					NotCompleteCount--
				} else if reply.VoteGranted {
					AcceptList = append(AcceptList, id)
					AcceptCount++
					NotCompleteCount--
				} else if reply.Term > args.Term {
					rf.Lock("SwitchToCandicate 2")
					rf.SwitchToFollower(-1, reply.Term)
					rf.Unlock()
					m.Unlock()
					rf.persist()
					return
				} else {
					NotCompleteCount--
				}
				m.Unlock()
			}(&a[i], i)
		}
	}

	tryCount := 0
	for {
		m.Lock()
		if AcceptCount > len(rf.peers)/2 {
			rf.SwitchToLeader(a[0].Term)
			m.Unlock()
			return
		} else if AcceptCount+NotCompleteCount > len(rf.peers)/2 {
			m.Unlock()
			tryCount++
			if tryCount > 10 {
				return
			}
			time.Sleep(heartsbeatsDurtaion)
		} else {
			rf.Lock("SwitchToCandicate 3")
			rf.SwitchToFollower(-1, rf.currentTerm)
			rf.Unlock()
			m.Unlock()
			rf.persist()
			return
		}
	}
}
func (rf *Raft) SwitchToLeader(switchTerm int) {
	rf.Lock("SwitchToLeader")
	rf.PrintlnAlreadyLock(direction{-1, -1}, "SwitchToLeader")
	for i := 0; i < len(rf.peers); i++ {
		rf.nextIndex[i] = rf.GetLogLen() + 1
		rf.matchIndex[i] = 0
	}
	if switchTerm < rf.currentTerm {
		rf.Unlock()
		return
	}
	if rf.status == Candidate {
		rf.status = Leader
	} else if rf.status == Follower {
		rf.Unlock()
		return
	}
	rf.Unlock()

}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.Println(direction{-1, -1}, rf.me, "Start", command)
	index := -1
	term := -1
	isLeader := true
	// Your code here (2B).
	rf.Lock("Start")
	if rf.status != Leader {
		isLeader = false
		rf.Unlock()
	} else {
		l := Log{rf.currentTerm, command}
		rf.log = append(rf.log, l)
		index = rf.GetLogLen()
		term = rf.currentTerm
		rf.matchIndex[rf.me] = rf.GetLogLen()
		rf.Unlock()
		rf.persist()
	}

	return index, term, isLeader
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}
func (rf *Raft) leaderSendAppendEntries() {
	// termNow := rf.currentTerm
	// okCount := 1
	if rf.killed() {
		return
	}
	for i, p := range rf.peers {
		if i != rf.me {
			go func(id int, peer *labrpc.ClientEnd) {
				rf.Lock("leaderSendAppendEntries")
				if rf.status != Leader {
					rf.Unlock()
					return
				}
				PrevLogIndex := rf.nextIndex[id] - 1
				var a []Log
				var PrevLogTerm int
				if PrevLogIndex > 0 {
					prevLog := rf.GetLog(PrevLogIndex)
					if prevLog == nil {
						// fmt.Println("prevLogIndex = ", PrevLogIndex, "prevLog is nil")
						go rf.LeaderInstallSnapshot(id)
						rf.Unlock()
						return
					}
					PrevLogTerm = prevLog.Term
					a = rf.GetLogRange(PrevLogIndex+1, rf.GetLogLen())
					if a == nil && PrevLogIndex+1 < rf.GetLogLen() {
						// fmt.Println("prevLogIndex = ", PrevLogIndex, "a is nil")
						go rf.LeaderInstallSnapshot(id)
						rf.Unlock()
						return
					}
				} else {
					PrevLogTerm = 0
					a = rf.log
				}
				args := AppendEntriesArgs{rf.currentTerm, rf.me, PrevLogIndex, PrevLogTerm, a, rf.commitIndex}
				if len(args.Entries) > 200 {
					args.Entries = args.Entries[:200]
				}
				reply := AppendEntriesReply{}
				rf.Unlock()
				if rf.killed() {
					return
				}
				ok := peer.Call("Raft.AppendEntries", &args, &reply)
				if ok {
					rf.Lock()
					if rf.status != Leader || args.Term != rf.currentTerm {
						rf.Unlock()
						return
					}
					if reply.Term > rf.currentTerm {
						rf.SwitchToFollower(-1, reply.Term)
					} else if reply.Success == false {
						// rf.nextIndex[id] = (rf.nextIndex[id] + rf.matchIndex[id]) / 2
						if reply.CorrectPrevIndex >= 0 && reply.CorrectPrevIndex < args.PrevLogIndex {
							rf.nextIndex[id] = reply.CorrectPrevIndex + 1
						} else {
							rf.nextIndex[id]--
						}

						if rf.nextIndex[id] <= 0 {
							rf.nextIndex[id] = 1
						}
					} else {
						// okCount++
						rf.matchIndex[id] = rf.nextIndex[id] - 1
						rf.nextIndex[id] += len(args.Entries)
						if rf.nextIndex[id] > rf.GetLogLen()+1 {
							rf.nextIndex[id] = rf.GetLogLen() + 1
						}
					}
					rf.Unlock()
					rf.persist()
				} else {
					rf.Println(direction{0, id}, "AppendEntries TimeOut")
					if !rf.killed() {
					}
				}

				// var match []int = make([]int, len(rf.peers))
				// copy(match, rf.matchIndex)
				// sort.Sort(match)

			}(i, p)
		}
	}
	time.Sleep(heartsbeatsDurtaion)
	rf.Lock("leaderSendAppendEntries 2")
	var commitCopy IntList = make(IntList, len(rf.peers))
	copy(commitCopy, rf.matchIndex)
	sort.Sort(commitCopy)
	if commitCopy[len(rf.peers)/2] > rf.commitIndex {
		if rf.GetLog(rf.GetLogLen()).Term == rf.currentTerm {
			rf.commitIndex = commitCopy[len(rf.peers)/2]
			go rf.ApplyChanges(rf.commitIndex)
		}
	}
	rf.Unlock()
}

type IntList []int

func (s IntList) Len() int           { return len(s) }
func (s IntList) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
func (s IntList) Less(i, j int) bool { return s[i] < s[j] }

// The ticker go routine starts a new election if this peer hasn't received
// heartsbeats recently.
func (rf *Raft) ApplyChanges(commitIndex int) {
	// rf.Lock("ApplyChanges")
	rf.muApplyChange.Lock()
	if rf.lastApplied < commitIndex {
		rf.Println(direction{1, -1}, "Applied Msg to", commitIndex, "Log end = ", rf.GetLog(commitIndex))
	}
	for ; rf.lastApplied < commitIndex; rf.lastApplied++ {
		msg := ApplyMsg{true, rf.GetLog(rf.lastApplied + 1).Log, rf.lastApplied + 1, false, []byte{}, 0, 0}
		// rf.PrintlnAlreadyLock(direction{1, -1}, "Applied Msg", msg)
		rf.applyCh <- msg
	}
	rf.Println(direction{1, -1}, "Applied Msg End")
	rf.muApplyChange.Unlock()
}
func (rf *Raft) ticker() {
	for rf.killed() == false {
		rf.Lock("ticker")
		status := rf.status
		lastOpTime := rf.lastOpTime
		if rf.commitIndex > rf.GetLogLen() {
		}
		rf.Unlock()
		if status == Leader {
			go rf.leaderSendAppendEntries()
		} else {
			if time.Now().After(lastOpTime.Add(startnewtermDuation)) {
				go rf.SwitchToCandicate()
				var t time.Duration = time.Duration(rand.Intn(500000000) * int(time.Nanosecond))
				t = t + electionDuationMin
				time.Sleep(t)
			} else {
				// rf.Lock("ticker 2")
				// status_cache_temp := rf.status
				// rf.Unlock()
				// if status_cache_temp == Candidate {
				// }

			}

		}
		time.Sleep(heartsbeatsDurtaion)

	}

	// Your code here to check if a leader election should
	// be started and to randomize sleeping time using
	// time.Sleep().

}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).
	rf.status = Follower
	rf.currentTerm = 0
	rf.votedFor = -1
	rf.log = make([]Log, 0)
	// rf.persist()
	rf.commitIndex = 0
	rf.lastApplied = 0
	rf.nextIndex = make([]int, len(rf.peers))
	rf.matchIndex = make([]int, len(rf.peers))
	rf.lastOpTime = time.Now()
	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	rf.applyCh = applyCh

	// start ticker goroutine to start elections
	go rf.ticker()

	return rf
}

type direction struct {
	send int /*0 for receive, 1 for send, -1 for self*/
	id   int
}

func TimeStr(time.Time) string {
	output := fmt.Sprintf("%02d:%03d", time.Now().Second(), time.Now().Nanosecond()/1000000)
	return output
}
func (rf *Raft) Println(d direction, a ...any) {
	rf.Lock("Println")
	defer rf.Unlock()
	rf.PrintlnAlreadyLock(d, a)
}
func (rf *Raft) PrintlnAlreadyLock(d direction, a ...any) {
	output := fmt.Sprint(rf.me, "[", TimeStr(time.Now()), ",", rf.currentTerm, "]")
	if rf.killed() {
		return
	}
	if d.send == 1 {
		if d.id != -1 {
			output = fmt.Sprint(output, rf.me, "->", d.id)
		} else {
			output = fmt.Sprint(output, rf.me, "->", "Client")
		}

	} else if d.send == 0 {
		if d.id != -1 {
			output = fmt.Sprint(output, rf.me, "<-", d.id)
		} else {
			output = fmt.Sprint(output, rf.me, "<-", "Client")
		}

	} else {
		output = fmt.Sprint(output, rf.me, ":")
	}
	// fmt.Println(output, a)
}

func (rf *Raft) GetLog(index int) *Log {
	if index == rf.snapShot.index {
		return &Log{rf.snapShot.term, nil}
	} else if index < rf.snapShot.index {
		// s := fmt.Sprintf("Get Invalid Index Value, index = %v, rf.snapShot.index = %v\n", index, rf.snapShot.index)
		// panic(s)
		return nil
	} else if index > rf.GetLogLen() {
		s := fmt.Sprintf("Invalid index = %v, the Max Value is %v", index, rf.GetLogLen())
		panic(s)
	}
	if index-1-rf.snapShot.index < 0 {
		return nil
	}
	return &rf.log[index-1-rf.snapShot.index]
}

func (rf *Raft) GetLogRange(indexBegin int, indexEnd int) []Log {
	/*Include Begin and End*/
	if indexEnd > rf.GetLogLen() {
		indexEnd = rf.GetLogLen()
	}
	if indexBegin > indexEnd {
		return nil
	}
	indexBegin -= rf.snapShot.index
	indexEnd -= rf.snapShot.index
	indexBegin--
	if indexBegin < 0 {
		return nil
	}
	return rf.log[indexBegin:indexEnd]
}

func (rf *Raft) TruncLog(index int) {
	/*all log after(not include) index are removed, index begin from 1*/
	rf.log = rf.log[:index-rf.snapShot.index]
}

func (rf *Raft) GetLogLen() int {
	return len(rf.log) + rf.snapShot.index
}

func (rf *Raft) PrintAllLogs() {
	var line1 string = "Index"
	var line2 string = "Value"
	var line3 string = "Term"
	for id, v := range rf.log {
		line1 = fmt.Sprintf("%s\t%v", line1, id+rf.snapShot.index+1)
		line2 = fmt.Sprintf("%s\t%v", line2, v.Log)
		line3 = fmt.Sprintf("%s\t%v", line3, v.Term)
	}
	fmt.Printf("%v's Log:\n%s\n%s\n%s\n", rf.me, line1, line2, line3)

}
